package streaming.demo.mq.kafka.config;

import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;

public class MySchema implements KeyedSerializationSchema {

    //element: 具体数据

    /**
     * 要发送的key
     *
     * @param element 原数据
     * @return key.getBytes
     */
    @Override
    public byte[] serializeKey(Object element) {
        //这里可以随便取你想要的key，然后下面分区器就根据这个key去决定发送到kafka哪个分区中，
        System.out.println("serializeKey:" + element.toString());
        System.out.println("serializeKey2:" + element.toString().split(" ")[0]);
        //element就是flink流中的真实数据，取出key后要转成字节数组
        return element.toString().split(" ")[0].getBytes();
    }

    /**
     * 要发送的value
     *
     * @param element 原数据
     * @return value.getBytes
     */
    @Override
    public byte[] serializeValue(Object element) {
        //要序列化的value，这里一般就原封不动的转成字节数组就行了
        System.out.println("serializeValue:" + element.toString());
        return element.toString().getBytes();
    }

    @Override
    public String getTargetTopic(Object element) {
        //这里返回要发送的topic名字，没什么用，可以不做处理
        return null;
    }

}
